package com.ttech.twitter.stream;

import com.mongodb.*;
import com.ttech.twitter.bean.TweetDbObject;
import com.ttech.twitter.util.DbUtil;
import com.ttech.twitter.util.GeoUtil;
import com.ttech.twitter.util.TweetUtil;
import org.apache.commons.lang.StringEscapeUtils;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.ObjectWriter;
import twitter4j.*;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * User: TTACANSEVER
 */
public class TwitterStreamer {

    private LinkedBlockingDeque<String> QUEUE = new LinkedBlockingDeque<String>(10000);

    private static final Set<String> medicalKeySet = new HashSet<String>();
    private static final Set<String> transportKeySet = new HashSet<String>();
    private static final Set<String> serviceKeySet = new HashSet<String>();


    static {
        medicalKeySet.add("doktor");
        medicalKeySet.add("hem\u015fir");
        medicalKeySet.add("hemsir");
        medicalKeySet.add("hastane");
        medicalKeySet.add("eczane");
        medicalKeySet.add("sa\u011flik");
        medicalKeySet.add("saglik");
        medicalKeySet.add("sa\u011fl\u0131k");
        medicalKeySet.add("ilac");
        medicalKeySet.add("ila\u00E7");
        medicalKeySet.add("hekim");
        medicalKeySet.add("hasta");
        medicalKeySet.add("cerrah");
        medicalKeySet.add("veteriner");

        transportKeySet.add("metrob\u00fcs");
        transportKeySet.add("metrobus");
        transportKeySet.add("metro");
        transportKeySet.add("tramvay");
        transportKeySet.add("taksi");
        transportKeySet.add("tren");
        transportKeySet.add("vapur");
        transportKeySet.add("trafik");
        transportKeySet.add("dolmu\u015E");
        transportKeySet.add("dolmus");
        transportKeySet.add("minibüs");
        transportKeySet.add("kopru");
        transportKeySet.add("k\u00F6pr\u00FC");
        transportKeySet.add("iskele");
        transportKeySet.add("\u015Fof\u00F6r");
        transportKeySet.add("ido");
        transportKeySet.add("iett");
        transportKeySet.add("bisiklet");
        transportKeySet.add("motor");
        transportKeySet.add("kamyon");
        transportKeySet.add("kamyonet");
        transportKeySet.add("araba");
        transportKeySet.add("otomobil");

        serviceKeySet.add("dogalgaz");
        serviceKeySet.add("do\u011Ealgaz");
        serviceKeySet.add("dogal gaz");
        serviceKeySet.add("do\u011Eal gaz");
        serviceKeySet.add("igda\u015F");

        serviceKeySet.add("elektrik");
        serviceKeySet.add("ayedas");
        serviceKeySet.add("ayeda\u015F");
        serviceKeySet.add("tedas");
        serviceKeySet.add("teda\u015F");
        serviceKeySet.add("bedas");
        serviceKeySet.add("beda\u015F");

        serviceKeySet.add("iski");
        serviceKeySet.add("kaldirim");
        serviceKeySet.add("kald\u0131r\u0131m");
        serviceKeySet.add("yol");
        serviceKeySet.add("sokak");
        serviceKeySet.add("\u00E7ukur");
        serviceKeySet.add("cukur");
        serviceKeySet.add("kaz\u0131");
        serviceKeySet.add("belediye");
        serviceKeySet.add("ibb");
        serviceKeySet.add("b\u00FCy\u00FCk\u015Fehir");
    }

    public void init() throws IOException {

        StatusListener listener = new StatusListener() {
            public void onStatus(Status status) {

                String type = "default";
                if(status.getText().toLowerCase().contains("sorunistanbul")) {
                    type = "problem";
                } else if(TweetUtil.sentenceContainsWord(status.getText(), medicalKeySet)) {
                    type = "medical";
                } else if(TweetUtil.sentenceContainsWord(status.getText(), transportKeySet)) {
                    type = "transportation";
                } else if(TweetUtil.sentenceContainsWord(status.getText(), serviceKeySet)) {
                    type = "service";
                }

                if (status.getGeoLocation() != null) {
                    if (GeoUtil.inBoundingBox(status.getGeoLocation().getLatitude(), status.getGeoLocation().getLongitude(), 40.778738, 28.26071, 41.22026, 29.559845)) {
                        try {
                            String tweet = status.getId()
                                    + "|||" + status.getGeoLocation().getLatitude()
                                    + "|||" + status.getGeoLocation().getLongitude()
                                    + "|||" + StringEscapeUtils.escapeHtml(status.getText())
                                    + "|||" + status.getUser().getScreenName()
                                    + "|||" + StringEscapeUtils.escapeHtml(status.getUser().getProfileImageURL())
                                    + "|||" + status.getUser().getProfileBannerURL()
                                    + "|||" + type;
                            QUEUE.put(tweet);

                            try {
                                TweetDbObject dbObject = new TweetDbObject(
                                        status.getId(),
                                        status.getUser().getScreenName(),
                                        status.getText(),
                                        status.getGeoLocation().getLatitude(),
                                        status.getGeoLocation().getLongitude(),
                                        status.getUser().getProfileImageURL(),
                                        type,
                                        new Timestamp(status.getCreatedAt().getTime()));
                                DbUtil.insertTweet(dbObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }

                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                } else if(status.getUser().getLocation() != null) {
                    if(status.getUser().getLocation().toLowerCase().contains("istanbul")
                            || status.getUser().getLocation().toUpperCase().contains("\u0130STANBUL")) {
                        TweetDbObject dbObject = new TweetDbObject(
                                status.getId(),
                                status.getUser().getScreenName(),
                                status.getText(),
                                0,
                                0,
                                status.getUser().getProfileImageURL(),
                                type,
                                new Timestamp(status.getCreatedAt().getTime()));
                        try {
                            DbUtil.insertTweet(dbObject);
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                }

            }

            public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
                //
            }

            public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
                //
            }

            @Override
            public void onScrubGeo(long l, long l2) {
                //
            }

            @Override
            public void onStallWarning(StallWarning stallWarning) {
                //
            }

            public void onException(Exception ex) {
                ex.printStackTrace();
            }
        };

        TwitterStream twitterStream = new TwitterStreamFactory().getInstance();
        twitterStream.addListener(listener);
        FilterQuery filter = new FilterQuery();
        //double[][] bBox = new double[][]{{28.372190, 40.7969165}, {29.68780376, 41.378610}};
        double[][] bBox = new double[][]{{25.000, 35.819}, {44.850, 42.100}};
        //double[][] bBox = new double[][]{{-124.7625, 24.5210}, {-66.9326, 49.3845}};
        filter.locations(bBox);
        //filter.track(new String[]{"mutluyum", "mutsuzum", "heyecanl\u0131y\u0131m", "k\u0131zg\u0131n\u0131m", "sinirliyim", "yorgunum"});
        //filter.track(new String[]{"a"});
        twitterStream.filter(filter);
        //twitterStream.sample();

    }

    public String getTweet() throws IOException, InterruptedException {
        do {

        } while (QUEUE.size() < 3);
        return QUEUE.take();
    }

}
